package deepbrain.simnetwork.task;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

import deepbrain.simnetwork.engine.SimEngine;
import deepbrain.simnetwork.exception.SimNetworkRuntimeException;
import deepbrain.simnetwork.monitor.Monitor;

/**
 * 
 * 
 * @author Li Yang
 * @create 2009-2-24
 */
public abstract class Task implements
		RunnableFuture<Map<String, Serializable>>, Serializable {

	private static final long serialVersionUID = -1805966149103321514L;

	protected ConcurrentHashMap<String, Serializable> taskResults;

	protected HashMap<String, Monitor<?>> monitors;

	protected String taskName;

	protected ThreadLocal<SimEngine> threadLocalEngine = new ThreadLocal<SimEngine>();

	private final Sync sync;

	public Task() {
		taskResults = new ConcurrentHashMap<String, Serializable>();
		monitors = new HashMap<String, Monitor<?>>();
		sync = new Sync(
				new ConcurrentTaskAdapter<ConcurrentHashMap<String, Serializable>>(
						this, taskResults));
	}

	public String getTaskName() {
		if (taskName != null)
			return taskName;
		else
			return "Task@" + hashCode();
	}

	public void setTaskName(String taskName) {
		this.taskName = taskName;
	}

	@Override
	public String toString() {
		return getTaskName();
	}

	public void setThreadLocalEngine(SimEngine engine) {
		this.threadLocalEngine.set(engine);
	}

	public SimEngine getThreadLocalEngine() {
		return this.threadLocalEngine.get();
	}

	public void addTaskResult(String key, Serializable value) {
		taskResults.put(key, value);
	}

	@SuppressWarnings("unchecked")
	public <T extends Serializable> T getTaskResult(String resultName) {
		return (T) taskResults.get(resultName);
	}

	public void addMonitor(Monitor<?> monitor) {
		monitors.put("" + monitor.hashCode(), monitor);
	}

	public void addMonitor(String monitorName, Monitor<?> monitor) {
		monitors.put(monitorName, monitor);
	}

	@SuppressWarnings("unchecked")
	public <T extends Task> Monitor<T> getMonitor(String name) {
		return (Monitor<T>) monitors.get(name);
	}

	public Monitor<?>[] getMonitors() {
		return monitors.values().toArray(new Monitor[0]);
	}

	public void removeMonitor(String name) {
		monitors.remove(name);
	}

	public void removeMonitor(Monitor<?> monitor) {
		monitors.values().remove(monitor);
	}

	/*
	 * @SuppressWarnings("unchecked") protected void fireBeforeTaskStart() { for
	 * (Monitor monitor : monitors.values()) { monitor.beforeTaskStart(this); }
	 * }
	 */
	@SuppressWarnings("unchecked")
	protected void fireTaskStarting() {
		for (Monitor monitor : getMonitors()) {
			monitor.onTaskStart(this);
		}
	}

	@SuppressWarnings("unchecked")
	protected void fireStepFinish() {
		for (Monitor monitor : getMonitors()) {
			monitor.onStepFinish(this);
		}
	}

	@SuppressWarnings("unchecked")
	protected void fireTaskFinish() {
		for (Monitor monitor : getMonitors()) {
			monitor.onTaskFinish(this);
			if (monitor.useOnce())
				removeMonitor(monitor);
		}
	}

	public boolean isCancelled() {
		return sync.innerIsCancelled();
	}

	public boolean isDone() {
		return sync.innerIsDone();
	}

	public boolean cancel(boolean mayInterruptIfRunning) {
		return sync.innerCancel(mayInterruptIfRunning);
	}

	public ConcurrentHashMap<String, Serializable> get()
			throws InterruptedException, ExecutionException {
		return sync.innerGet();
	}

	public ConcurrentHashMap<String, Serializable> get(long timeout,
			TimeUnit unit) throws InterruptedException, ExecutionException,
			TimeoutException {
		return sync.innerGet(unit.toNanos(timeout));
	}

	public boolean isRunning() {
		return sync.innerIsRunning();
	}

	public void run() {
		try {
			fireTaskStarting();
			sync.innerRun();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * Subclasses must override this method rather than run()
	 */
	public abstract void runTask();

	private void done() {
		try {
			fireTaskFinish();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private final class Sync extends AbstractQueuedSynchronizer {
		private static final long serialVersionUID = 9040260548104452047L;
		/** State value representing that task is running */
		private static final int RUNNING = 1;
		/** State value representing that task ran */
		private static final int RAN = 2;
		/** State value representing that task was cancelled */
		private static final int CANCELLED = 4;

		/** The underlying callable */
		private final Callable<ConcurrentHashMap<String, Serializable>> callable;
		/** The result to return from get() */
		private ConcurrentHashMap<String, Serializable> result;
		/** The exception to throw from get() */
		private Throwable exception;

		/**
		 * The thread running task. When nulled after set/cancel, this indicates
		 * that the results are accessible. Must be volatile, to ensure
		 * visibility upon completion.
		 */
		private volatile Thread runner;

		Sync(Callable<ConcurrentHashMap<String, Serializable>> callable) {
			this.callable = callable;
		}

		private boolean ranOrCancelled(int state) {
			return (state & (RAN | CANCELLED)) != 0;
		}

		/**
		 * Implements AQS base acquire to succeed if ran or cancelled
		 */
		protected int tryAcquireShared(int ignore) {
			return innerIsDone() ? 1 : -1;
		}

		/**
		 * Implements AQS base release to always signal after setting final done
		 * status by nulling runner thread.
		 */
		protected boolean tryReleaseShared(int ignore) {
			runner = null;
			return true;
		}

		boolean innerIsCancelled() {
			return getState() == CANCELLED;
		}

		boolean innerIsDone() {
			return ranOrCancelled(getState()) && runner == null;
		}

		boolean innerIsRunning() {
			return getState() == RUNNING;
		}

		ConcurrentHashMap<String, Serializable> innerGet()
				throws InterruptedException, ExecutionException {
			acquireSharedInterruptibly(0);
			if (getState() == CANCELLED)
				throw new CancellationException();
			if (exception != null)
				throw new ExecutionException(exception);
			return result;
		}

		ConcurrentHashMap<String, Serializable> innerGet(long nanosTimeout)
				throws InterruptedException, ExecutionException,
				TimeoutException {
			if (!tryAcquireSharedNanos(0, nanosTimeout))
				throw new TimeoutException();
			if (getState() == CANCELLED)
				throw new CancellationException();
			if (exception != null)
				throw new ExecutionException(exception);
			return result;
		}

		void innerSet(ConcurrentHashMap<String, Serializable> v) {
			for (;;) {
				int s = getState();
				if (ranOrCancelled(s))
					return;
				if (compareAndSetState(s, RAN))
					break;
			}
			result = v;
			releaseShared(0);
			done();
		}

		void innerSetException(Throwable t) {
			for (;;) {
				int s = getState();
				if (ranOrCancelled(s))
					return;
				if (compareAndSetState(s, RAN))
					break;
			}
			exception = t;
			result = null;
			releaseShared(0);
			done();
			throw new SimNetworkRuntimeException(t);
		}

		boolean innerCancel(boolean mayInterruptIfRunning) {
			for (;;) {
				int s = getState();
				if (ranOrCancelled(s))
					return false;
				if (compareAndSetState(s, CANCELLED))
					break;
			}
			if (mayInterruptIfRunning) {
				Thread r = runner;
				if (r != null)
					r.interrupt();
			}
			releaseShared(0);
			done();
			return true;
		}

		void innerRun() {
			if (!compareAndSetState(0, RUNNING))
				return;
			try {
				runner = Thread.currentThread();
				innerSet(callable.call());
			} catch (Throwable ex) {
				innerSetException(ex);
			}
		}

		boolean innerRunAndReset() {
			if (!compareAndSetState(0, RUNNING))
				return false;
			try {
				runner = Thread.currentThread();
				callable.call(); // don't set result
				runner = null;
				return compareAndSetState(RUNNING, 0);
			} catch (Throwable ex) {
				innerSetException(ex);
				return false;
			}
		}
	}

	static final class ConcurrentTaskAdapter<T> implements Callable<T> {
		final Task task;
		final T result;

		ConcurrentTaskAdapter(Task task, T result) {
			this.task = task;
			this.result = result;
		}

		public T call() {
			task.runTask();
			return result;
		}
	}
}
